package rabbitmq

import (
	"fmt"
	"github.com/streadway/amqp"
	"os"
	"path/filepath"
	"reflect"
	"strconv"
	"strings"
	c "xrt/constants"
	"xrt/core"
	"xrt/utils"
	"xrt/utils/file/log"
)

type rabbitMqLater struct {
	host           string
	port           string
	vHost          string
	user           string
	password       string
	listenQueue    string
	pushQueue      string
	noAck          bool
	prefetchCount  int
	contentType    string
	separator      byte
	commandSign    string
	commandLowCase bool
	conn           *amqp.Connection
	channel        *amqp.Channel
	queueListen    *amqp.Queue
	queuePush      *amqp.Queue
	patternMap     map[string]map[string]func(*RbChannel)
}

func (r *rabbitMqLater) Initial(configs map[string]string, asyncMaps map[string]interface{}) {
	//初始化
	r.host = configs["host"]
	r.port = configs["port"]
	r.vHost = configs["vhost"]
	r.user = configs["user"]
	r.password = configs["password"]
	r.listenQueue = configs["listen_queue"]
	r.pushQueue = configs["push_queue"]
	r.noAck, _ = strconv.ParseBool(configs["no_ack"])
	r.prefetchCount = utils.ParseInt(configs["prefetch_count"])
	r.contentType = configs["content_type"]
	r.separator = configs["separator"][0]
	r.commandSign = configs["command_sign"]
	r.commandLowCase = utils.ParseBool(configs["command_low_case"])
	connectStr := fmt.Sprintf("amqp://%s:%s@%s:%s%s", r.user, r.password, r.host, r.port, r.vHost)
	if core.IsDebug() {
		log.Debugf("Initial RabbitMQLayer: [%s]\n", connectStr)
	}
	//初始化连接
	if conn, err := amqp.Dial(connectStr); err != nil {
		panic(fmt.Sprintf("RabbitMQ Initial Failed, [%s].", err.Error()))
	} else {
		r.conn = conn
		//初始化通道
		if ch, err := conn.Channel(); err != nil {
			panic(fmt.Sprintf("RabbitMQ Initial Failed, [%s].", err.Error()))
		} else {
			r.channel = ch
			r.channel.Qos(1, 0, false)
			//初始化监听队列
			if q1, err := r.channel.QueueDeclare(
				r.listenQueue, // name
				true,          // durable
				false,         // delete when unused
				false,         // exclusive
				false,         // no-wait
				nil,           // arguments
			); err != nil {
				panic(fmt.Sprintf("RabbitMQ Initial Failed, [%s].", err.Error()))
			} else {
				r.queueListen = &q1
			}
			//初始化推送队列
			if q2, err := r.channel.QueueDeclarePassive(
				r.pushQueue, // name
				true,        // durable
				false,       // delete when unused
				false,       // exclusive
				false,       // no-wait
				nil,         // arguments
			); err != nil {
				panic(fmt.Sprintf("RabbitMQ Initial Failed, [%s].", err.Error()))
			} else {
				r.queuePush = &q2
			}
		}
	}
	//生成Mux路由
	r.patternMap = map[string]map[string]func(*RbChannel){}
	for pattern, handler := range asyncMaps {
		muxTyp := reflect.TypeOf(handler)
		muxVal := reflect.ValueOf(handler)
		//将AppHead的根目录设置为默认根路径
		if strings.HasPrefix(muxTyp.String(), "*"+c.AppHead+".") {
			pattern = fmt.Sprintf("%s", strings.SplitN(filepath.Clean(pattern), string(os.PathSeparator), 3)[2])
		} else {
			pattern = pattern[1:]
		}
		//设置所有命名为Default的RouteSync为默认访问路径
		if strings.HasSuffix(pattern, "Default") {
			pattern = pattern[0 : len(pattern)-8]
		}
		pattern = strings.Replace(pattern, string(os.PathSeparator), string(r.separator), -1)
		//命令是否全部为小写解析
		if r.commandLowCase {
			pattern = strings.ToLower(pattern)
		}
		r.patternMap[pattern] = map[string]func(*RbChannel){}
		for i := 0; i < muxTyp.NumMethod(); i++ {
			muxTypMethod := muxTyp.Method(i)
			muxValMethod := muxVal.Method(i)
			methodName := muxTypMethod.Name
			if r.commandLowCase {
				methodName = strings.ToLower(methodName)
			}
			r.patternMap[pattern][methodName] = muxValMethod.Interface().(func(*RbChannel))
			//如果是默认方法, 则打印生成日志时不需要输出方法名
			asyncRoute := fmt.Sprintf("%s%c%s", pattern, r.separator, methodName)
			if strings.ToLower(methodName) == "default" {
				asyncRoute = asyncRoute[:len(asyncRoute)-8]
			}
			log.Debugf("Generate Async Route Pattern: %s", asyncRoute)
		}
	}
}

func (r *rabbitMqLater) Listening() {
	defer r.conn.Close()
	defer r.channel.Close()

	forever := make(chan bool)

	if listenChannel, err := r.channel.Consume(
		r.queueListen.Name, // queue
		"",                 // consumer
		false,              // no-ack
		false,              // exclusive
		false,              // no-local
		false,              // no-wait
		nil,                // args
	); err != nil {
		panic(fmt.Sprintf("RabbitMQ Listening Failed, [%s].", err.Error()))
	} else {
		go func() {
			for d := range listenChannel {
				log.Debugf("Received Async message: %s", d.Body)
				rb := RbChannel{r, &d, "", nil}
				//无法解析出命令则忽略当前指令
				if !rb.Analysis() {
					log.Debugf("Null Command, Ignore It.")
					continue
				}
				go func() {
					dfCmd := "Default"
					if r.commandLowCase {
						dfCmd = strings.ToLower(dfCmd)
					}

					//直接访问default方法
					if _, ok := r.patternMap[rb.Command]; ok {
						if _, okDefault := r.patternMap[rb.Command][dfCmd]; okDefault {
							r.patternMap[rb.Command][dfCmd](&rb)
						} else {
							log.Debugf("Command [%s] Doesn't Have Default Function.", rb.Command)
						}
						//分类下的其他方法
					} else {
						spPath := strings.SplitN(rb.Command, string(r.separator), -1)
						pattern := strings.Join(spPath[:len(spPath)-2], string(r.separator))
						methods := spPath[len(spPath)-1]
						if r.commandLowCase {
							methods = strings.ToLower(methods)
						}

						if _, okPattern := r.patternMap[pattern]; okPattern {
							if _, okMethod := r.patternMap[pattern][methods]; okMethod {
								r.patternMap[pattern][methods](&rb)
							} else {
								log.Errorf("Command [%s] Doesn't Exists.", rb.Command)
							}
						} else {
							log.Errorf("Command [%s] Doesn't Exists.", rb.Command)
						}
					}
					//是否需要确认
					if !r.noAck {
						r.channel.Ack(d.DeliveryTag, false)
					}
				}()
			}
		}()
	}

	log.Infof(fmt.Sprintf("Start RabbitMQ Comsuming [%s:%s] ...", r.host, r.port))
	<-forever
}

func (r *rabbitMqLater) Push(msg string) {
	r.channel.Publish(
		"", r.queuePush.Name, false, false,
		amqp.Publishing{
			ContentType:  "text/plain",
			DeliveryMode: amqp.Persistent, Body: []byte(msg)})
}

func (r *rabbitMqLater) PushSelf(msg string) {
	r.channel.Publish(
		"", r.queueListen.Name, false, false,
		amqp.Publishing{
			ContentType:  "text/plain",
			DeliveryMode: amqp.Persistent, Body: []byte(msg)})
}

func init() {
	core.X.RegisterComponent(c.ImplAsync, rabbitMqLater{})
}
